package com.unism.uniMeterage.meterage.stream.version1_1;

import com.unism.uniMeterage.meterage.stream.*;
import com.unism.uniMeterage.meterage.stream.version1_1.util.PrivateUtil;

import java.io.FileInputStream;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 系统默认管理者
 */
public class DefaultManager implements Manager{
    //worker缓存,用于管理
    private static ConcurrentHashMap<String,Worker> CacheMapWorker=new ConcurrentHashMap<String,Worker>();
    private static final String WorkerName="Worker_Name";
    private static final String MaxMessageNum="Max_Message_Num";
    private static final  String ThreadGroupName="DefaultManagerThreadGroup";
    private static final  String DefaultSubject="DefaultSubject";
    private static final String  ConfigPath="";
    private Object managerMonitor=new Object();//用于对CacheMapWorker进行增删时对排他锁（9成情况用不到）
    private ThreadGroup threadGroup;
    private static  DefaultManager instance;
    public static DefaultManager getDefaultInstance() {
        if (instance == null) {
            instance = new DefaultManager();
            Map<String, Object> options = null;
            Properties p = new Properties();
            try {
                p.load(new FileInputStream(DefaultManager.class.getClassLoader().getResource("manager.properties").getPath()));
            }catch(IOException e) {
                try{
                    p.load(new FileInputStream("/resource/manager.properties"));
                }catch (IOException e1) {
                    System.out.println("DefaultManager: no configutration: "+e);
                }
            }
            if (!PrivateUtil.isEmpty(p)) {
                options = new HashMap<String, Object>();
                for (Object key : p.keySet()) {
                    String skey = (String) key;
                    options.put(skey, p.getProperty(skey));
                }
            }
            instance.initialize(options);
        }
        return instance;
    }

    /**
     * 动态生成指定的Worker
     * @param clazz
     * @param name
     * @param options
     * @return
     */
    @Override
    public Worker createWorker(Class<?> clazz, String name, Map<String, Object> options) {
        AbstractWorker worker = null;
        synchronized (managerMonitor) {
            if (!CacheMapWorker.containsKey(name)) {
                try {
                    worker= (AbstractWorker) clazz.newInstance();
                    String workerName=options.get(WorkerName).toString();
                    Integer maxMessageNum=Integer.valueOf(options.get(MaxMessageNum).toString());
                    worker.initWorker(workerName,this,getThreadGroup(), maxMessageNum);
                } catch (Exception e) {
                    throw new IllegalArgumentException("create worker happend error " + name,e);
                }
            } else {
                throw new IllegalArgumentException("name already in use: " + name);
            }
        }
        return worker;
    }
    @Override
    public Worker createAndStartWorker(Class<?> clazz, String name, Map<String, Object> options) {
        AbstractWorker worker=(AbstractWorker)this.createWorker(clazz,name,options);
        startWorker(worker);
        worker.start();
        return worker;
    }
    @Override
    public void startWorker(Worker worker) {
        worker.activate();
    }

    @Override
    public void deleteWorker(Worker worker) {
        if(worker != null){
            if(worker.getWorkerName() != null){
                CacheMapWorker.remove(worker.getWorkerName());
            }
        }
    }
    @Override
    public Integer sendMessage(Message message, Worker from, Worker to) {
        int count = 0;
        if (message != null) {
            AbstractWorker worker= (AbstractWorker) to;
            if (worker != null) {
                DefaultMessage defaultMessage = (DefaultMessage)message;
                //添加默认属性
                if(message.getSource() == null){
                    defaultMessage.setSource(from);
                }
                if(message.getSubject() == null){
                    defaultMessage.setSubject(DefaultSubject);
                }
                switch (worker.currStatus()){
                    case NEW:
                        System.out.println("存在异常的Worker状态");
                        break;
                    case WORKING:
                    case WAITING:
                        worker.addMessage(message);
                        defaultMessage.triggerListeners(new MessageEvent(from,defaultMessage, MessageEvent.MessageStatus.SENT));
                        count++;
                       // worker.notifyWorker();
                        break;
                    case DESTORY:
                        System.out.println("不处理Destory状态Worker");
                        break;
                    default:
                        break;
                }
            }
        }
        return count;
    }
    @Override
    public Integer sendMessage(Message message, Worker from, String category) {
        //保留
        if(CacheMapWorker.containsKey(category)){
            this.sendMessage(message,from,CacheMapWorker.get(category));
        }
         return null;
    }
    @Override
    public Integer getWorkerCount(Class clazz) {
        int res = 0;
        if (clazz != null) {
            synchronized (managerMonitor) {
                for (String key : CacheMapWorker.keySet()) {
                    Worker worker = CacheMapWorker.get(key);
                    if (clazz.isAssignableFrom(worker.getClass())) {
                        res++;
                    }
                }
            }
        }
        return res;
    }

    @Override
    public Boolean existWorker(String workerName) {
        if(CacheMapWorker !=null){
            return CacheMapWorker.containsKey(workerName);
        }
        return false;
    }
    @Override
    public void initialize(Map<String, Object> options) {
        System.out.println("初始化 manager 开始");
        try{
            for (Map.Entry<String,Object> entry : options.entrySet()){
                String keyClazz=entry.getKey();
                if(PrivateUtil.isMatch(PrivateUtil.getDefaultKeyClazz(),keyClazz)){
                    String keyName=PrivateUtil.getKeyName(keyClazz);
                    String keyNumber=PrivateUtil.getKeyNum(keyClazz);
                    Object valueName=options.get(keyName);
                    Object valueNumber=options.get(keyNumber);
                    Object valueClazz=options.get(keyClazz);
                    Integer workerNum =0;
                    if(keyName == null || valueName == null){
                        throw new IllegalArgumentException("worker name is invalid ===>"+valueClazz);
                    }
                    if(keyClazz == null || valueClazz == null){
                        throw new IllegalArgumentException("worker class is invalid ===>"+valueClazz);
                    }
                    if(keyNumber == null || valueNumber == null){
                        workerNum=1;
                    }else{
                        workerNum=Integer.valueOf(valueNumber.toString());
                    }
                    HashMap opt=new HashMap();
                    opt.put(WorkerName,valueName);
                    opt.put(MaxMessageNum,99999);
                    Worker worker=this.createAndStartWorker(Class.forName(valueClazz.toString()),valueName.toString(),opt);
                    Message message=new DefaultMessage(worker,"liuc",1);
                    this.sendMessage(message,worker,worker);
                }
            }
        }catch(Exception e){
          System.out.println("manager 初始化失败"+e.getMessage());
        }
        System.out.println("初始化 manager 结束");
    }
    @Override
    public void terminateAndWait() {
        //保留
    }
    @Override
    public void terminate() {
        //保留
    }
    private  ThreadGroup getThreadGroup(){
        if(threadGroup == null){
            threadGroup=new ThreadGroup(ThreadGroupName);
        }
        return threadGroup;
    }
}
